package som.simon.reactor.sinks;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import som.simon.reactor.SimonUtil;

import java.util.stream.IntStream;

public class SinksMultiMain {

    public static void main(String[] args) {

        // Sinks ： 接收器

        // Sinks.UnicastSpec unicast = Sinks.many().unicast();   // 单播     ：只能绑定单个订阅者
        // Sinks.MulticastSpec multicast = Sinks.many().multicast(); // 多播
        // Sinks.MulticastReplaySpec replay = Sinks.many().replay();  // 重播

        Sinks.Many<Integer> sink = Sinks.many().multicast()
                .onBackpressureBuffer();

        // 异步发消息
        new Thread(() -> {
            IntStream.rangeClosed(1, 10).forEach(ix -> {
                sink.tryEmitNext(ix);
                SimonUtil.sleep();
            });
        }).start();

        Flux<Integer> flux = sink.asFlux();
        flux.map(SimonUtil::format)
                .subscribe(v -> System.out.println("v1 = " + v));

        // 异步二次订阅
        new Thread(() -> {
            // 等待5秒
            SimonUtil.sleep(5);
            flux.subscribe(
                    v -> System.out.println("v2 = " + v),
                    System.out::println,
                    () -> System.out.println("game over....")
            );
        }).start();

        SimonUtil.pause();
    }

    /**
     * 输出结果：
     * v1 = 0000001
     * v1 = 0000002
     * v1 = 0000003
     * v1 = 0000004
     * v1 = 0000005
     * v1 = 0000006
     * v1 = 0000007
     * TODO v2 = 7
     * v1 = 0000008
     * v2 = 8
     * v1 = 0000009
     * v2 = 9
     * v1 = 0000010
     * v2 = 10
     *
     * TODO 结论：订阅者默认从开始订阅那一刻起接受元素
     */

}
